package com.itc.bi.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itc.bi.dto.MqttClientDto;
import com.itc.bi.entity.DirectoryEntity;
import com.itc.bi.entity.MqMqttClientEntity;
import com.itc.bi.mqtt.MqttClientCallback;
import com.itc.bi.mqtt.MqttClientConnect;
import com.itc.bi.service.DirectoryService;
import com.itc.bi.service.MqMqttClientService;
import com.itc.bi.mapper.MqMqttClientMapper;
import com.itc.bi.utils.PaginationUtils;
import com.itc.common.core.exception.ServiceException;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author songxiude
 * @description 针对表【mq_mqtt_client(mqtt链接信息表)】的数据库操作Service实现
 * @createDate 2024-07-15 14:36:47
 */
@Service
@Slf4j
public class MqMqttClientServiceImpl extends ServiceImpl<MqMqttClientMapper, MqMqttClientEntity>
        implements MqMqttClientService {

    @Autowired
    private DirectoryService directoryService;

    /**
     * 添加修改mqtt链接
     *
     * @param mqMqttClient
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean mqttSave(MqMqttClientEntity mqMqttClient) {
        //判断链接是否已存在
        long count = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, mqMqttClient.getClientId()));
        if (ObjectUtils.isEmpty(mqMqttClient.getId()) && count > 0) {
            throw new ServiceException("该链接已存在不可重复创建");
        }
        //判断数据集是不是目录
        DirectoryEntity directoryEntity = directoryService.getOne(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId, mqMqttClient.getDtId()));
        if (ObjectUtils.isEmpty(directoryEntity) || directoryEntity.getIsDir().intValue() != 0) {
            throw new ServiceException("数据集不是目录不能添加新的链接");
        }
        //组装链接地址前缀
        if (mqMqttClient.getClientType().intValue() == 1) {
            mqMqttClient.setMqttPrefix("tcp://");
        } else {
            mqMqttClient.setMqttPrefix("ssl://");
        }
        //如果不传客户端链接id自动生成
        if (StringUtil.isEmpty(mqMqttClient.getClientId())) {
            mqMqttClient.setClientId(IdWorker.get32UUID());
        }
        //链接mqtt
        this.initMqttServer(mqMqttClient);
        mqMqttClient.setStatus(1);
        return save(mqMqttClient);
    }

    /**
     * 修改mqtt链接信息
     *
     * @param mqMqttClient
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean mqttUpdate(MqMqttClientEntity mqMqttClient) {
        //先停掉运行中的链接
        ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
        MqttClientConnect clientConnect = mqttClients.get(mqMqttClient.getClientId());
        if (ObjectUtils.isEmpty(clientConnect)) {
            try {
                clientConnect.close();
            } catch (MqttException e) {
                log.error("关闭链接失败");
                throw new ServiceException("关闭链接失败");
            }
        }
        //组装链接地址前缀
        if (mqMqttClient.getClientType().intValue() == 1) {
            mqMqttClient.setMqttPrefix("tcp://");
        } else {
            mqMqttClient.setMqttPrefix("ssl://");
        }
        //链接mqtt
        this.initMqttServer(mqMqttClient);
        mqMqttClient.setStatus(1);
        return updateById(mqMqttClient);
    }

    /**
     * 获取mqtt链接信息
     *
     * @param mqttClientDto
     * @return
     */
    public IPage<MqMqttClientEntity> mqttList(MqttClientDto mqttClientDto) {
        Page<MqMqttClientEntity> page = PaginationUtils.<MqMqttClientEntity>getPage(mqttClientDto.getPage(), mqttClientDto.getSize());
        LambdaQueryWrapper<MqMqttClientEntity> wrapper = Wrappers.<MqMqttClientEntity>lambdaQuery();
        //排除部分敏感字段
        wrapper.select(MqMqttClientEntity.class, client -> !client.getColumn().equals("password"));
        //wrapper.select(MqMqttClientEntity::getId,MqMqttClientEntity::getMqttName);
        wrapper.like(StringUtil.isNotEmpty(mqttClientDto.getMqttName()), MqMqttClientEntity::getMqttName, mqttClientDto.getMqttName());
        return page(page, wrapper);
    }

    /**
     * 手动启动mqtt
     *
     * @param id
     * @return
     */
    public boolean startMqtt(Long id) {
        MqMqttClientEntity clientEntity = getById(id);
        if (ObjectUtils.isEmpty(clientEntity)) {
            throw new ServiceException("不存在的实例");
        }
        if (clientEntity.getStatus().intValue() == 1) {
            throw new ServiceException("已经正常链接请不要频繁重连");
        }
        boolean b = this.initMqttServer(clientEntity);
        if (b) {
            clientEntity.setStatus(1);
        }
        updateById(clientEntity);
        return b;
    }

    /**
     * 删除mqtt链接
     *
     * @param clientId
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean removeMqtt(String clientId) {
        try {
            ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
            MqttClientConnect mqttClientConnect = mqttClients.get(clientId);
            if(ObjectUtil.isEmpty(mqttClientConnect)){
                return false;
            }
            mqttClientConnect.close();
            mqttClients.remove(clientId);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("断开连接失败,id--->" + clientId);
            return false;
        }
        return remove(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
    }

    /**
     * 初始化链接mqtt
     *
     * @param mqMqttClient
     */
    public boolean initMqttServer(MqMqttClientEntity mqMqttClient) {
        try {
            MqttClientConnect mqttClientConnect = new MqttClientConnect();
            mqttClientConnect.setMqttClientId(mqMqttClient.getClientId());
            mqttClientConnect.setMqttClient(mqMqttClient, new MqttClientCallback(mqMqttClient.getClientId()));
            mqttClientConnect.sub(StringUtils.commaDelimitedListToStringArray(mqMqttClient.getClientTopic()));
            MqttClientConnect.mqttClients.put(mqMqttClient.getClientId(), mqttClientConnect);
            log.info("mqtt初始化链接成功,host--->" + mqMqttClient.getServerHost());
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("新增连接失败,host--->" + mqMqttClient.getServerHost());
            return false;
        }
    }


}




